use crate::cmp::{Cmp, MemtableKeyCmp};
use crate::key_types::{build_memtable_key, parse_internal_key, parse_memtable_key, ValueType};
use crate::key_types::{LookupKey, UserKey};
use crate::skipmap::{SkipMap, SkipMapIter};
use crate::types::{current_key_val, LdbIterator, SequenceNumber};

use std::rc::Rc;

use bytes::Bytes;
use integer_encoding::FixedInt;

/// Provides Insert/Get/Iterate, based on the SkipMap implementation.
/// MemTable uses MemtableKeys internally, that is, it stores key and value in the [Skipmap] key.
///
/// 这段代码定义了一个名为 MemTable 的公共结构体，它是一个基于跳表（SkipMap）实现的内存表，提供插入（Insert）、获取（Get）和迭代（Iterate）功能。
pub struct MemTable {
    /// MemTable 内部包含一个 SkipMap 类型的字段 map，说明它使用跳表作为底层数据结构来存储数据
    /// 注释提到 MemTable 在内部使用 MemtableKeys，这意味着它将键（key）和值（value）都存储在跳表的键（key）中，这是一种特殊的存储方式（通常跳表的每个节点会分别存储键和值）
    map: SkipMap,
}

impl MemTable {
    /// Returns a new MemTable.
    /// This wraps opt.cmp inside a MemtableKey-specific comparator.
    pub fn new(cmp: Rc<Box<dyn Cmp>>) -> MemTable {
        MemTable::new_raw(Rc::new(Box::new(MemtableKeyCmp(cmp))))
    }

    /// Doesn't wrap the comparator in a MemtableKeyCmp.
    fn new_raw(cmp: Rc<Box<dyn Cmp>>) -> MemTable {
        MemTable {
            map: SkipMap::new(cmp),
        }
    }

    pub fn len(&self) -> usize {
        self.map.len()
    }

    /// 返回当前 跳表 所使用的近似内存量（以字节为单位）
    pub fn approx_mem_usage(&self) -> usize {
        self.map.approx_memory()
    }

    /// 这个 add 方法是内存表（MemTable）的核心写入接口，用于将用户的键值对（或删除操作）按照内存表的内部格式，插入到底层的数据结构中。
    ///
    /// seq: SequenceNumber：唯一序列号（用于版本控制，序列号越大数据越新）；
    ///
    /// t: ValueType：操作类型（TypeValue 表示正常数据，TypeDeletion 表示删除标记）；
    ///
    /// key: UserKey<'_>：用户传入的原始键；
    ///
    /// value: &[u8]：用户传入的原始值（删除操作时通常为空）。
    pub fn add(&mut self, seq: SequenceNumber, t: ValueType, key: UserKey<'_>, value: &[u8]) {
        self.map.insert(build_memtable_key(key, value, t, seq), Vec::new())
    }

    /// get returns the value for the given entry and whether the entry is marked as deleted. This
    /// is to distinguish between not-found and found-deleted.
    ///
    /// 这个 get 方法是内存表（MemTable）的查询接口，用于根据查找键（LookupKey）从内存表中获取对应的值，同时判断该键是否被标记为删除。
    /// 它是 LSM 树存储引擎中内存层数据查询的核心实现。
    ///
    /// 这个 get 方法的核心功能是实现内存表的高效查询，并处理两种特殊情况：
    ///
    /// 键存在且有值：返回对应的值；
    ///
    /// 键存在但被删除：返回 None 但标记为已删除（用于后续合并 SSTable 时清理数据）；
    ///
    /// 键不存在：返回 None 且标记为未删除。
    #[allow(unused_variables)]
    pub fn get(&self, key: &LookupKey) -> (Option<Bytes>, bool) {
        // self.map 是内存表的底层数据结构（通常是有序映射，如跳表或平衡树）；
        // seek(key.memtable_key()) 根据内存表中存储的键格式（memtable_key，包含用户键、序列号和类型）定位到目标位置。
        let mut iter = self.map.iter();
        // 将迭代器定位到跳表中大于或等于目标键（key） 的第一个节点位置。
        iter.seek(key.memtable_key());
        // 在seek方法中已经找到了数据, 并且已经修改了current指针, 现在从 跳表 SkipMapIter 中直接解析索引块, 拿到数据就可以了
        if let Some((foundkey, _)) = current_key_val(&iter) {
            // 解析找到的键（提取用户键长度、偏移量、类型标记等）
            let (fkeylen, fkeyoff, tag, vallen, valoff) = parse_memtable_key(&foundkey);

            // Compare user key -- if equal, proceed
            // We only care about user key equality here
            // 比较用户键（忽略序列号和类型，只关注实际的用户键是否相等）
            if key.user_key() == &foundkey[fkeyoff..fkeyoff + fkeylen] {
                // 类型为普通值：返回值的字节切片（转换为 Bytes 类型）
                // tag & 0xff 是一个位运算操作，用于获取 tag 变量的最低 8 位（一个字节）的值。
                // 具体解释：
                // 0xff 是十六进制表示，对应的二进制是 11111111（8 个 1）
                // & 是按位与运算，规则是：只有当两个操作数对应位都为 1 时，结果位才为 1，否则为 0
                // 当 tag 与 0xff 进行按位与时，会保留 tag 最低 8 位的原始值，而更高位的所有位都会被清零。
                // 举例来说：
                // 若 tag 是 0x12345678（二进制 00010010 00110100 01010110 01111000）
                // 则 tag & 0xff 的结果是 0x78（二进制 01111000），即最低 8 位的值
                if tag & 0xff == ValueType::TypeValue as u64 {
                    // println!("foundkey[valoff..valoff + vallen].to_vec().into(): {:?}", foundkey[valoff..valoff + vallen].to_vec().into());
                    return (
                        Some(foundkey[valoff..valoff + vallen].to_vec().into()),
                        false,
                    );
                } else {
                    // 类型为删除标记：返回 None，但标记为已删除（true）
                    return (None, true);
                }
            }
        }
        (None, false)
    }

    pub fn iter(&self) -> MemtableIterator {
        MemtableIterator {
            skipmapiter: self.map.iter(),
        }
    }
}

/// MemtableIterator is an iterator over a MemTable. It is mostly concerned with converting to and
/// from the MemtableKey format used in the inner map; all key-taking or -returning methods deal
/// with InternalKeys.
///
/// This iterator does not skip deleted entries.
pub struct MemtableIterator {
    skipmapiter: SkipMapIter,
}

impl LdbIterator for MemtableIterator {
    fn advance(&mut self) -> bool {
        println!("----------mem advance----------");
        if !self.skipmapiter.advance() {
            return false;
        }
        self.skipmapiter.valid()
    }
    fn reset(&mut self) {
        println!("----------mem reset----------");
        self.skipmapiter.reset();
    }
    fn prev(&mut self) -> bool {
        println!("----------mem prev----------");
        // Make sure this is actually needed (skipping deleted values?).
        loop {
            if !self.skipmapiter.prev() {
                return false;
            }
            if let Some((key, _val)) = self.skipmapiter.current() {
                let (_, _, tag, _, _) = parse_memtable_key(&key);

                if tag & 0xff == ValueType::TypeValue as u64 {
                    return true;
                } else {
                    continue;
                }
            } else {
                return false;
            }
        }
    }
    fn valid(&self) -> bool {
        println!("----------mem valid----------");
        self.skipmapiter.valid()
    }
    /// current returns the current key (in InternalKey format) and value.
    fn current(&self) -> Option<(Bytes, Bytes)> {
        println!("----------mem current----------");
        if !self.valid() {
            return None;
        }

        if let Some((key_bytes, _val_bytes)) = self.skipmapiter.current() {
            let (keylen, keyoff, _, vallen, valoff) = parse_memtable_key(&key_bytes);
            let internal_key =
                Bytes::copy_from_slice(&key_bytes[keyoff..keyoff + keylen + u64::required_space()]);
            let value = Bytes::copy_from_slice(&key_bytes[valoff..valoff + vallen]);
            Some((internal_key, value))
        } else {
            None
        }
    }
    /// seek takes an InternalKey.
    fn seek(&mut self, to: &[u8]) {
        println!("----------mem seek----------");
        // Assemble the correct memtable key from the supplied InternalKey.
        let (_, seq, ukey) = parse_internal_key(to);
        self.skipmapiter
            .seek(LookupKey::new(ukey, seq).memtable_key());
    }
}

/// shift_left moves s[mid..] to s[0..s.len()-mid]. The new size is s.len()-mid.
fn shift_left(s: &mut Vec<u8>, mid: usize) {
    for i in mid..s.len() {
        s.swap(i, i - mid);
    }
    let newlen = s.len() - mid;
    s.truncate(newlen);
}

#[cfg(test)]
#[allow(unused_variables)]
mod tests {
    use super::*;
    use crate::key_types::{parse_tag, truncate_to_userkey};
    use crate::options;
    use crate::test_util::{test_iterator_properties, LdbIteratorIter};

    #[test]
    fn test_shift_left() {
        let mut v = vec![1, 2, 3, 4, 5];
        shift_left(&mut v, 1);
        assert_eq!(v, vec![2, 3, 4, 5]);

        let mut v = vec![1, 2, 3, 4, 5];
        shift_left(&mut v, 4);
        assert_eq!(v, vec![5]);
    }

    fn get_memtable() -> MemTable {
        let mut mt = MemTable::new(options::for_test().cmp);
        let entries = vec![
            (ValueType::TypeValue, 115, "abc", "122"),
            (ValueType::TypeValue, 120, "abc", "123"),
            (ValueType::TypeValue, 121, "abd", "124"),
            (ValueType::TypeDeletion, 122, "abe", "125"),
            (ValueType::TypeValue, 123, "abf", "126"),
        ];

        for e in entries.iter() {
            mt.add(e.1, e.0, e.2.as_bytes(), e.3.as_bytes());
        }
        mt
    }

    #[test]
    fn test_memtable_parse_tag() {
        let tag = (12345 << 8) | 1;
        assert_eq!(parse_tag(tag), (ValueType::TypeValue, 12345));
    }

    #[test]
    fn test_memtable_add() {
        let mut mt = MemTable::new(options::for_test().cmp);
        mt.add(
            123,
            ValueType::TypeValue,
            "abc".as_bytes(),
            "123".as_bytes(),
        );

        assert_eq!(
            mt.map.iter().next().unwrap().0,
            &[11, 97, 98, 99, 1, 123, 0, 0, 0, 0, 0, 0, 3, 49, 50, 51]
        );
        assert_eq!(
            mt.iter().next().unwrap().0,
            &[97, 98, 99, 1, 123, 0, 0, 0, 0, 0, 0]
        );
    }

    #[test]
    fn test_memtable_add_get() {
        let mt = get_memtable();

        // Smaller sequence number doesn't find entry
        if let Some(v) = mt.get(&LookupKey::new("abc".as_bytes(), 110)).0 {
            eprintln!("{:?}", v);
            panic!("found");
        }

        if let Some(v) = mt.get(&LookupKey::new("abf".as_bytes(), 110)).0 {
            eprintln!("{:?}", v);
            panic!("found");
        }

        // Bigger sequence number falls back to next smaller
        if let Some(v) = mt.get(&LookupKey::new("abc".as_bytes(), 116)).0 {
            assert_eq!(v, "122".as_bytes());
        } else {
            panic!("not found");
        }

        // Exact match works
        if let (Some(v), deleted) = mt.get(&LookupKey::new("abc".as_bytes(), 120)) {
            assert_eq!(v, "123".as_bytes());
            assert!(!deleted);
        } else {
            panic!("not found");
        }

        if let (None, deleted) = mt.get(&LookupKey::new("abe".as_bytes(), 122)) {
            assert!(deleted);
        } else {
            panic!("found deleted");
        }

        if let Some(v) = mt.get(&LookupKey::new("abf".as_bytes(), 129)).0 {
            assert_eq!(v, "126".as_bytes());
        } else {
            panic!("not found");
        }
    }

    #[test]
    fn test_memtable_iterator_init() {
        let mt = get_memtable();
        let mut iter = mt.iter();

        assert!(!iter.valid());
        iter.next();
        assert!(iter.valid());
        assert_eq!(
            current_key_val(&iter).unwrap().0,
            vec![97, 98, 99, 1, 120, 0, 0, 0, 0, 0, 0].as_slice()
        );
        iter.reset();
        assert!(!iter.valid());
    }

    #[test]
    fn test_memtable_iterator_seek() {
        let mt = get_memtable();
        let mut iter = mt.iter();

        assert!(!iter.valid());

        iter.seek(LookupKey::new("abc".as_bytes(), 400).internal_key());
        let (mut gotkey, gotval) = current_key_val(&iter).unwrap();
        truncate_to_userkey(&mut gotkey);
        assert_eq!(
            ("abc".as_bytes(), "123".as_bytes()),
            (gotkey.as_slice(), gotval.as_slice())
        );

        iter.seek(LookupKey::new("xxx".as_bytes(), 400).internal_key());
        assert!(!iter.valid());

        iter.seek(LookupKey::new("abd".as_bytes(), 400).internal_key());
        let (mut gotkey, gotval) = current_key_val(&iter).unwrap();
        truncate_to_userkey(&mut gotkey);
        assert_eq!(
            ("abd".as_bytes(), "124".as_bytes()),
            (gotkey.as_slice(), gotval.as_slice())
        );
    }

    #[test]
    fn test_memtable_iterator_fwd() {
        let mt = get_memtable();
        let mut iter = mt.iter();

        let expected = [
            "123".as_bytes(), /* i.e., the abc entry with
                               * higher sequence number comes first */
            "122".as_bytes(),
            "124".as_bytes(),
            // deleted entry:
            "125".as_bytes(),
            "126".as_bytes(),
        ];
        for (i, (k, v)) in LdbIteratorIter::wrap(&mut iter).enumerate() {
            assert_eq!(v, expected[i]);
        }
    }

    #[test]
    fn test_memtable_iterator_reverse() {
        let mt = get_memtable();
        let mut iter = mt.iter();

        // Bigger sequence number comes first
        iter.next();
        assert!(iter.valid());
        assert_eq!(
            current_key_val(&iter).unwrap().0,
            vec![97, 98, 99, 1, 120, 0, 0, 0, 0, 0, 0].as_slice()
        );

        iter.next();
        assert!(iter.valid());
        assert_eq!(
            current_key_val(&iter).unwrap().0,
            vec![97, 98, 99, 1, 115, 0, 0, 0, 0, 0, 0].as_slice()
        );

        iter.next();
        assert!(iter.valid());
        assert_eq!(
            current_key_val(&iter).unwrap().0,
            vec![97, 98, 100, 1, 121, 0, 0, 0, 0, 0, 0].as_slice()
        );

        iter.prev();
        assert!(iter.valid());
        assert_eq!(
            current_key_val(&iter).unwrap().0,
            vec![97, 98, 99, 1, 115, 0, 0, 0, 0, 0, 0].as_slice()
        );

        iter.prev();
        assert!(iter.valid());
        assert_eq!(
            current_key_val(&iter).unwrap().0,
            vec![97, 98, 99, 1, 120, 0, 0, 0, 0, 0, 0].as_slice()
        );

        iter.prev();
        assert!(!iter.valid());
    }

    #[test]
    fn test_memtable_parse_key() {
        let key = vec![11, 1, 2, 3, 1, 123, 0, 0, 0, 0, 0, 0, 3, 4, 5, 6];
        let (keylen, keyoff, tag, vallen, valoff) = parse_memtable_key(&key);
        assert_eq!(keylen, 3);
        assert_eq!(&key[keyoff..keyoff + keylen], vec![1, 2, 3].as_slice());
        assert_eq!(tag, (123 << 8) | 1);
        assert_eq!(vallen, 3);
        assert_eq!(&key[valoff..valoff + vallen], vec![4, 5, 6].as_slice());
    }

    #[test]
    fn test_memtable_iterator_behavior() {
        let mut mt = MemTable::new(options::for_test().cmp);
        let entries = [
            (115, "abc", "122"),
            (120, "abd", "123"),
            (121, "abe", "124"),
            (123, "abf", "126"),
        ];

        for e in entries.iter() {
            mt.add(e.0, ValueType::TypeValue, e.1.as_bytes(), e.2.as_bytes());
        }

        test_iterator_properties(mt.iter());
    }
}
